package com.xahj.one

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * @author dongtangqiang
 */
abstract class SparkJobTemplate {
  //1. 创建RDD

  //2. RDD的transformation操作
  //   一个RDD调用转换算子，生成另外一个RDD

  //3. RDD的action操作
  //   一个RDD调用行动算子，生成另外一个非RDD的对象，或者直接将结果输出到外部存储

  //4. RDD的持久化
  //   缓存cache persist
  //   检查点checkpoint

  /**
   * 创建程序入口类的实例
   *
   * @param conf 传入创建使用的配置文件对象
   * @return
   */
  def getSC(conf: SparkConf) = {
    new SparkContext(conf)
  }


  /**
   * 创建程序入口类的实例
   *
   * @param conf 传入创建使用的配置文件对象
   * @return
   */
  def getSession(args: Array[String]) = {
    val builder = SparkSession.builder()
    if (args.length == 0) {
      builder.master("local[2]")
        .appName("demo01")
    }
    val spark = builder
      .getOrCreate()
    spark
  }

  /**
   * 创建程序入口类的实例
   *
   * @param conf 传入创建使用的配置文件对象
   * @return
   */
  def getSC(args: Array[String]) = {
    val conf = new SparkConf()
    conf.setAppName(this.getClass.getName)
    if (args.length == 0) {
      conf.setMaster("local[2]")
    }
    new SparkContext(conf)
  }

  /**
   * 执行程序的主逻辑,封装转换操作
   *
   * @param rdd 输入数据
   * @tparam T 输入数据类型
   * @tparam U 输出数据的类型
   * @return
   */
  def execute[T, U](rdd: RDD[T]): RDD[U]

  /**
   * 执行任务结束的数据输出，封装行动操作
   *
   * @param rdd     最终需要输出的数据集
   * @param outPath 如果需要输出到文件，可以指定输出路径
   * @tparam T 最终需要输出的数据类型
   */
  def finish[T](rdd: RDD[T], outPath: String = "")
}
